package org.example.client.verii;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.example.client.veri.bind.Client;
import org.example.client.veri.bind.ServiceProxyFactory;
import org.example.codec.RpcDecoder;
import org.example.codec.RpcEncoder;
import org.example.common.RpcRequest;
import org.example.common.RpcResponse;

import java.net.InetSocketAddress;

/**
 * @author mojito
 * @Classname Client
 * @Description TODO
 * @Date 2020/7/20 下午9:46
 **/
@Slf4j
public class RpcProxyNioClient extends SimpleChannelInboundHandler<RpcResponse> implements Client {
    private String host;
    private int port;
    private RpcResponse response;
    private Object lock = new Object();

    public RpcProxyNioClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public <T> T getService(Class<T> clazz) {
        ServiceProxyFactory<T> proxyFactory = new ServiceProxyFactory<>(clazz);
        return proxyFactory.newInstance(this);
    }

    @Override
    public RpcResponse process(RpcRequest rpcRequest) {
        NioEventLoopGroup executors = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(executors)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(host, port))
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        //in
                        pipeline.addLast(new RpcDecoder(RpcResponse.class));
                        pipeline.addLast(new RpcEncoder(RpcRequest.class));
                        pipeline.addLast(RpcProxyNioClient.this);
                        //out
                    }
                }).option(ChannelOption.TCP_NODELAY, true);
        ;
        ChannelFuture sync = null;
        try {
            sync = bootstrap.connect().sync();
            sync.channel().writeAndFlush(rpcRequest).sync();
            synchronized (lock) {
                lock.wait();
            }
            if (rpcRequest != null) {
                sync.channel().closeFuture();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            executors.shutdownGracefully();
        }
        return response;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcResponse rpcResponse) throws Exception {
        this.response = rpcResponse;
        synchronized (lock) {
            lock.notifyAll();
        }
        log.info("rpc response: [{}]", rpcResponse);
    }
}
